package com.fdb.smcts.core.threadpool;

import java.util.Collection;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class ApplicationThreadPool implements ThreadPool {
	
	private String poolName;
	
	//运行状态
	private boolean isRunning = false;
	
	//待处理队列
	private BlockingQueue<Callable<ExecuteRtnInfo>> taskQueue;// = new LinkedBlockingQueue<Callable<ExecuteRtnInfo>>();
	
	private int queueSize;
	
	//已处理完成队列
	private BlockingQueue<Future<ExecuteRtnInfo>> finishQueue;
	
	//执行任务队列数量
	private Map<String, Integer> execute_num = new HashMap<String, Integer>();

	//最大线程数量
	private volatile int maxTaskNum;
	
	//活动线程数量
	private Integer live_thread = 0;
	
	//成功数量
	private int success_num = 0;
	
	//执行失败数量
	private int failure_num = 0;
	
	//执行完成数量
	private int finished_task = 0;
	
	private Thread startThead;
	
	private static Logger logger = LoggerFactory.getLogger(ApplicationThreadPool.class);
	
	public void initThreadPool() {
		taskQueue = new LinkedBlockingQueue<Callable<ExecuteRtnInfo>>(queueSize);
		finishQueue = new LinkedBlockingQueue<Future<ExecuteRtnInfo>>(this.maxTaskNum + 1);
		startThead = new Thread(() -> start());
		startThead.start();
	}
	
	public void start() {
		
		if (!this.isRunning) {
			this.isRunning = true;
		}
		int i = 1;
		while (this.isRunning) {
			if (!this.taskQueue.isEmpty()) {
				try {
					
					while (this.live_thread >= this.maxTaskNum) {
						this.unblock();
						Thread.sleep(500);
						Thread.yield();
					}
					
//					synchronized (this.taskQueue) { // 无用的同步
					final Callable<ExecuteRtnInfo> callable = this.taskQueue.take();
					Future<ExecuteRtnInfo> future = Executors.newSingleThreadExecutor().submit(callable);
					String clazzName = callable.getClass().getName();
					if (!execute_num.containsKey(clazzName)) {
						execute_num.put(clazzName, 0);
					}
					execute_num.put(clazzName, execute_num.get(clazzName) + 1);
					synchronized(live_thread) {
						live_thread ++;
					}

					synchronized (this.finishQueue) {
						if (this.finishQueue.size() >= maxTaskNum - 1) {
							releaseFinishedThread(this.finishQueue.take());
						}
						this.finishQueue.put(future);
					}

					logger.info(this.toString());
//					}
				} catch (Throwable e) {
					logger.error("线程池运行线程异常!", e);
				}
			} else {
				try {
					Thread.sleep(200);
					if (this.taskQueue.isEmpty()) {
						this.unblock();
					}
				} catch (InterruptedException e) {
					logger.error("线程池释放任务异常!", e);
				}
				++i;
			}
			if (i % 1200 == 0) {
				logger.info(this.toString());
			}
		}
	
	}

	/**
	 * 当配置文件更改时调用的方法
	 * @param newQueueSize 新的队列大小
	 * @param newMaxTaskNum 新的最大线程数
	 */
	public void change(int newQueueSize, int newMaxTaskNum) {
		if (newMaxTaskNum < 0 || newMaxTaskNum < 0)
			throw new IllegalArgumentException();
		// 加锁，防止change期间taskQueue仍然有新任务
		synchronized(taskQueue) {
			// 首先更改最大任务数限制start线程创建新线程
			this.maxTaskNum = newMaxTaskNum;
			if (newQueueSize == this.queueSize) {
				return;
			}
			BlockingQueue<Callable<ExecuteRtnInfo>> newTaskQueue = new LinkedBlockingQueue<>(newQueueSize);
			while (this.taskQueue.size() > 0) {
				// 自旋等待start线程将taskQueue现存的任务派发，直到taskQueue的任务数量降到newQueueSize
			}
			this.taskQueue = newTaskQueue;
		}
	}
	
	@Deprecated
	public void stop() {
		this.unblock();
		this.isRunning = false;
		try {
			synchronized (taskQueue) {
				if (taskQueue.isEmpty()) {
					this.startThead.stop();
				} else {
					Thread.sleep(1000);
					Thread.yield();
				}
			}
		} catch (Exception e) {
			logger.error("停止线程池异常!", e);
		}
	}
	
	public void setPoolName(String name) {
		this.poolName = name;
	}
	
	public void unblock() {
		synchronized (this.finishQueue) {
			for (Future<ExecuteRtnInfo> future : this.finishQueue) {
				this.releaseFinishedThread(future);
				this.finishQueue.remove(future);
			}
			this.finishQueue.notify();
		}
	}
	
	public void releaseFinishedThread(Future<ExecuteRtnInfo> future) {
		try { 
			if(future.isDone()) {
				this.finished_task++;
				ExecuteRtnInfo info = future.get();
				switch (info.getRtn()) {
					case SUCCESS : this.success_num++; break;
					case EXCEPTION : 
						break;
					case FAILURE : 
						this.failure_num++; 
						break;
				}
				synchronized(this.live_thread) {
					live_thread--;
				}
				
				
			} else {
				this.finishQueue.add(future);
			}
		} catch (Exception e) {
			logger.error("线程池运行线程异常!", e);
        }
	}
	
	public void add(Callable<ExecuteRtnInfo> task) throws Exception {
		synchronized(taskQueue) {
			try {
				taskQueue.put(task);
			} catch (Exception e) {
				logger.error("任务："+ task +"加入线程池失败!", e);
				throw e;
			} finally {
				taskQueue.notify();
			}
		}
	}
	
	public void add(Collection<Callable<ExecuteRtnInfo>> tasks) {
		synchronized (taskQueue) {
			try {
				for (Callable<ExecuteRtnInfo> call : tasks) {
					taskQueue.put(call);
				}
			} catch (Exception e) {
				logger.error("任务："+ tasks +"加入线程池失败!", e);
			} finally {
				taskQueue.notify();
			}
			
		}
	}
	
	public void addExecuteNumber(String key) {
		if (!this.execute_num.containsKey(key)) {
			this.execute_num.put(key, 0);
		}
		this.execute_num.put(key, this.execute_num.get(key) + 1);
	}
	
	public boolean getRunning() {
		return this.isRunning;
	}
	
	public void setRunning(boolean isRunning) {
		if (isRunning) {
			if (!this.isRunning)
				this.isRunning = isRunning;
		}
	}
	
	@SuppressWarnings("deprecation")
	public void destroy() {
		synchronized (this) {
			this.toString();
			this.isRunning = false;
			this.taskQueue.clear();
			this.unblock();
			this.execute_num.clear();
			this.startThead.stop();
		}
	}

	public BlockingQueue<Callable<ExecuteRtnInfo>> getTaskQueue() {
		return taskQueue;
	}

	public void setTaskQueue(BlockingQueue<Callable<ExecuteRtnInfo>> taskQueue) {
		synchronized (this.taskQueue) {
			if (this.taskQueue.size() > 0) {
				this.taskQueue.addAll(taskQueue);
			}
			this.taskQueue.notify();
		}
	}
	
	public void addTaskQueue(BlockingQueue<Callable<ExecuteRtnInfo>> taskQueue) {
		synchronized (this.taskQueue) {
			this.taskQueue.addAll(taskQueue);
			this.taskQueue.notify();
		}
	}

	public BlockingQueue<Future<ExecuteRtnInfo>> getFinishQueue() {
		return finishQueue;
	}

	public void setFinishQueue(BlockingQueue<Future<ExecuteRtnInfo>> finishQueue) {
		this.finishQueue = finishQueue;
	}

	public Integer getMaxTaskNum() {
		return maxTaskNum;
	}

	public void setMaxTask(int maxTask) {
		synchronized (finishQueue) {
			BlockingQueue<Future<ExecuteRtnInfo>> temp = new LinkedBlockingQueue<Future<ExecuteRtnInfo>>();
			temp.addAll(this.finishQueue);
			this.maxTaskNum = maxTask;
			this.finishQueue = new LinkedBlockingQueue<Future<ExecuteRtnInfo>>(this.maxTaskNum + 1);
			this.finishQueue.addAll(temp);
			temp.clear();
			finishQueue.notify();
		}
	}

	public Integer getLiveThreadNum() {
		return live_thread;
	}

	public void addLiveThread(int liveThread) {
		if (this.live_thread == null) {
			this.live_thread = 0;
		}
		this.live_thread += liveThread;
	}

	public Map<String, Integer> getExecuteNum() {
		return execute_num;
	}

	public void setExecuteNum(Map<String, Integer> executeNum) {
		if (executeNum != null && executeNum.size() > 0) {
			for (String key : executeNum.keySet()) {
				if (!this.execute_num.containsKey(key)) {
					this.execute_num.put(key, executeNum.get(key));
				} else {
					this.execute_num.put(key, executeNum.get(key) + this.execute_num.get(key));
				}
			}
		}
	}

	public int getSuccess_num() {
		return success_num;
	}

	public void setSuccess_num(int successNum) {
		success_num = successNum;
	}

	public int getFailureNum() {
		return failure_num;
	}

	public void setFailureNum(int failure_num) {
		this.failure_num = failure_num;
	}

	public boolean isRunning() {
		return isRunning;
	}
	
	public String toString() {
		StringBuffer sb = new StringBuffer();
		sb.append("Pool name:"+ this.poolName);
		sb.append(" Queue number: "+ this.taskQueue.size());
		sb.append(" WorkThread number:" + this.maxTaskNum);
		sb.append(" LiveThread number:"+ this.live_thread);
		sb.append(" Finished task number:"+ this.finished_task);
		sb.append(" Success task number:"+ this.success_num);
		sb.append(" Failure task number:"+ this.failure_num);
    	sb.append(" Wait task number:" + this.taskQueue.size());
    	sb.append(" Execute number:"+ this.execute_num.toString());
		return sb.toString();
	}

	public void setMaxTaskNum(int maxTaskNum) {
		this.maxTaskNum = maxTaskNum;
	}

	public int getQueueSize() {
		return queueSize;
	}

	public void setQueueSize(int queueSize) {
		this.queueSize = queueSize;
	}

}
